/*
 * @version: 1.0
 * @Author: zhl
 * @Date: 2021-12-14 11:35:01
 * @LastEditors: zhanggq
 * @LastEditTime: 2023-11-30 15:33:33
 * @Description: 文件实现描述
 */

#include <typeinfo>
#include <sstream>

#include "ModbusMessageManager.h"
#include "CLogger.h"
#include "ModbusParseTask.h"
#include "CDBManager.h"
#include "common.h"
#include "hex.h"
#include "gw_manager.h"

#define MAX_REPONSE_LEN 2048

MessageManager MessageManager::c_mm;

#define save_log(data, data_len, direct) {                           \
            if((NULL != data) && (0 < data_len)){                    \
                int buf_len = data_len*2+1;                          \
                char buf[buf_len] = {0};                             \
                if(0 == hex_to_str(data, data_len, buf, &buf_len)){  \
                    INTERACTIVE_MESSAGE msg;                         \
                    msg.task_id = task_id;                           \
                    msg.direction = direct;                          \
                    msg.execution_time = get_timestamp();            \
                    msg.content = buf;                               \
                    interfacive_msg.msgs.push_back(msg);             \
                } \
            } \
        }



MessageManager::MessageManager(/* args */){
    c_req_id = 0;
    c_thread_pool.init(1);
}

MessageManager::~MessageManager(){
    c_thread_pool.destory();
}
//
MessageManager &MessageManager::getInstance(){
    return c_mm;
}

template<typename T>
T MessageManager::get_real_value(std::string formula, double scale, int decimal, std::string value){
    double v = std::stod(value);
    char ret_buf[64] = {0};
    char str_format[32] = {0};
    if(formula.empty()){
        if(1.0f != scale){
            v = v * scale;
        }
    } else {
        std::string strFormula = "echo \""+formula + "\"|bc";
        std::stringstream fstr;
        fstr << value;
        strFormula = string_replace(strFormula, "X", fstr.str());
        if(0 == exec_cmd((char *)strFormula.c_str(), ret_buf, sizeof(ret_buf))){
            v = std::stod(ret_buf);
        }
    }
    if(0 < decimal){
        snprintf(str_format, sizeof(str_format), "%%.%df", decimal);
        snprintf(ret_buf, sizeof(ret_buf), str_format, v);
        if(std::is_same<T,float>::value || std::is_same<T, double>::value){
            return std::stod(ret_buf);
        } else if(std::is_same<T,char>::value || std::is_same<T, short>::value || 
                std::is_same<T,int>::value || std::is_same<T,long>::value || 
                std::is_same<T, long long>::value){
            return (T)std::stoll(ret_buf);
        } else {
            return (T)std::stoull(ret_buf);
        }
    }else{
        return (T)v;
    }
    
    return (T)-999999;
}

std::string MessageManager::get_real_value_str(std::shared_ptr<MODBUS_COMMAND> modbus_control, std::string value, std::string& vstr){
    if((nullptr == modbus_control) || value.empty()){
        return "";
    }
    try{
        switch(modbus_control->data_type){
            case DATA_TYPE_BIT:{
                uint8_t  v = get_real_value<uint8_t>(modbus_control->formula, modbus_control->scale, modbus_control->decimal, value);
                vstr = std::to_string(v);
            }
            break;
            case DATA_TYPE_BITS:
                vstr = value;                
            break;
            case DATA_TYPE_SHORT:{
                short  v = get_real_value<short>(modbus_control->formula, modbus_control->scale, modbus_control->decimal, value);
                vstr = std::to_string(v);
            }
            break;
            case DATA_TYPE_UNSIGNED_SHORT:{
                uint16_t  v = get_real_value<uint16_t>(modbus_control->formula, modbus_control->scale, modbus_control->decimal, value);
                vstr = std::to_string(v);
            }
            case DATA_TYPE_INT:{
                int  v = get_real_value<int>(modbus_control->formula, modbus_control->scale, modbus_control->decimal, value);
                vstr = std::to_string(v);
            }
            break;
            case DATA_TYPE_UNSIGNED_INT:{
                uint32_t  v = get_real_value<uint32_t>(modbus_control->formula, modbus_control->scale, modbus_control->decimal, value);
                vstr = std::to_string(v);
            }
            break;
            case DATA_TYPE_FLOAT:{
                float  v = get_real_value<float>(modbus_control->formula, modbus_control->scale, modbus_control->decimal, value);
                vstr = std::to_string(v);
            }
            break;
            default:
            break;
        }
    } catch(...){
        CLogger::getInstance()->log_e("获取写数据异常.");
    }
    return vstr; 
}

int64_t MessageManager::get_req_id(){
    std::lock_guard<std::mutex> lock(c_mutex);
    c_req_id++;
    if(0 > c_req_id){
        c_req_id = 0;
    }
	return c_req_id;
}

int MessageManager::run_collect(std::shared_ptr<CConfigManager> cm, 
                                IChannel *channel_ptr, 
                                std::shared_ptr<CMosquittoManager> mm, 
                                TASK_PARAMS *dev_reqs, std::shared_ptr<TERMINAL_DEVICE> dev){
    if((NULL == cm) || (nullptr == channel_ptr) || (nullptr == dev_reqs)) {
        return -1;
    }
    std::list<std::shared_ptr<COLLECT_DATA>> data; 
    std::unordered_map<std::string, RESPONSE_DATA> resp_datas;
    for(auto item = dev_reqs->items.begin(); item != dev_reqs->items.end(); item++){        
        auto task_item = cm->get_item(item->item_id);
        RESPONSE_DATA resp_data;
        resp_data.point_id = -1;
        resp_data.code  = -1;
        resp_data.name  = std::to_string(item->item_id);
        resp_data.msg   = "";
        resp_data.value = "";
        if(nullptr == task_item){
            CLogger::getInstance()->log_w("不存在该任务项:%lld", item->item_id);
            resp_data.msg = "不存在该读取项";
            resp_datas.insert(std::make_pair(resp_data.name, resp_data));
            continue;
        }
        //
        std::shared_ptr<COLLECT_DATA> collect_data = std::make_shared<COLLECT_DATA>();
        int ret = read_handler(cm, channel_ptr, mm, dev, task_item->task_id, task_item->task_item_id, collect_data);

        if(0 > ret){
            resp_data.msg = "采集数据失败";
            resp_data.code = ret;
            collect_data = nullptr;
        }
        if(nullptr !=  collect_data){
            data.push_back(collect_data);
        } else {
            resp_datas.insert(std::make_pair(resp_data.name, resp_data));
        }
    }
    //所有串口请求完成后，提交给线程进行数据解析处理包括告警、本地存储等
    if(0 < data.size()){
        ModbusParseTask *pTask = new ModbusParseTask(cm, mm, dev_reqs, data, resp_datas);
        c_thread_pool.add_task(pTask);
    }
    return 0;
}
int MessageManager::run_read_cmd(std::shared_ptr<CConfigManager> cm, IChannel *channel_ptr, std::shared_ptr<CMosquittoManager> mm, TASK_PARAMS *dev_reqs, std::shared_ptr<TERMINAL_DEVICE> dev){
    if((NULL == cm) || (nullptr == channel_ptr) || (nullptr == dev_reqs) || (nullptr == dev)) {
        return -1;
    }
    std::list<std::shared_ptr<COLLECT_DATA>> data; 
    std::unordered_map<std::string, RESPONSE_DATA> resp_datas;
    for(auto item = dev_reqs->items.begin(); item != dev_reqs->items.end(); item++){        
        auto command = cm->get_modbus_command(item->item_id);
        RESPONSE_DATA resp_data;
        resp_data.point_id = -1;
        resp_data.code  = -1;
        resp_data.name  = std::to_string(item->item_id);
        resp_data.msg   = "";
        resp_data.value = "";
        if(nullptr == command){
            CLogger::getInstance()->log_w("不存在该读取项:%lld", item->item_id);
            resp_data.msg = "不存在该读取项";
            resp_datas.insert(std::make_pair(resp_data.name, resp_data));
            continue;
        }
        resp_data.name = command->name;
        //
        auto proto = cm->get_protocol_rule(command->protocol_id);
        if(nullptr == proto){
            continue;
        }
        std::shared_ptr<COLLECT_DATA> collect_data = std::make_shared<COLLECT_DATA>();
        int ret = read_cmd_handler(channel_ptr, dev, command, collect_data, proto->timeout);
        switch(ret){
            case 0:
                resp_data.msg = "采集数据返回成功";
                resp_data.code = ret;
            break;
            case -1:
                resp_data.msg = "采集数据返回失败(设备："+std::to_string(dev_reqs->dev_id)+")";
                resp_data.code = ret;
                collect_data = nullptr;
            break;
            case -2:
                resp_data.msg = "采集开关量数据失败";
                resp_data.code = ret;
                collect_data = nullptr;
            break;
            case -3:
                resp_data.msg = "采集信号量数据失败";
                resp_data.code = ret;
            break;
            case -10:
                resp_data.msg = "封装请求报文异常";
                resp_data.code = ret;
                collect_data = nullptr;
            break;
            case -200:
                resp_data.msg = "读取请求超时";
                resp_data.code = ret;
                collect_data = nullptr;
            break;
            case -201:
                resp_data.msg = "读取指令(未知功能码)";
                resp_data.code = ret;
                collect_data = nullptr;
            break;
            default:
                resp_data.code = -4;
                resp_data.msg = "解析数据异常";
                collect_data = nullptr;
            break;
        }
        resp_datas.insert(std::make_pair(resp_data.name, resp_data));
        if(nullptr != collect_data){
            data.push_back(collect_data);
        }
    }
    //所有串口请求完成后，提交给线程进行数据解析处理包括告警、本地存储等
    if(0 < data.size()){
        ModbusParseTask *pTask = new ModbusParseTask(cm, mm, dev_reqs, data, resp_datas);
        c_thread_pool.add_task(pTask);
    }
    return 0;
}
int MessageManager::run_write_cmd(std::shared_ptr<CConfigManager> cm, IChannel *channel_ptr, std::shared_ptr<CMosquittoManager> mm, TASK_PARAMS *dev_reqs, std::shared_ptr<TERMINAL_DEVICE> dev){
    if((NULL == cm) || (nullptr == channel_ptr) || (nullptr == dev_reqs) || (nullptr == dev)) {
        return -1;
    }
    int ret = 0;
    int r = 0;
    std::unordered_map<std::string, RESPONSE_DATA> resp_datas;
    std::unordered_map<std::string, RESPONSE_DATA> resp_update_datas;
    for(auto item = dev_reqs->items.begin(); item != dev_reqs->items.end(); item++){
        RESPONSE_DATA resp_data;
        resp_data.name = dev->name;
        resp_data.value = "";
        auto modbus_control = cm->get_modbus_command(item->item_id);
        if(nullptr == modbus_control){
            CLogger::getInstance()->log_w("不存在该指令规则:%lld", item->item_id);
            resp_data.msg  = "命令执行失败,不存在该指令规则";
            resp_data.code = -1;
            resp_data.point_id = -1;
            resp_data.reg_addr = -1;
            resp_datas.insert(std::make_pair("Invalid", resp_data));
            continue;
        }
        resp_data.reg_addr = modbus_control->reg_addr;
        resp_data.point_id = modbus_control->id;
        if(atoi(dev->slave_addr.c_str()) == 0xFF){//本机设备
            if(TASK_TYPE_SET_CMD_RULE == dev_reqs->type){
                resp_data.value = item->value;
                r = gw_write_handler(modbus_control, item->value);
            } else {
                resp_data.value = modbus_control->def_value;
                r = gw_write_handler(modbus_control, modbus_control->def_value);
            }
        } else {
            std::string vstr = "";
            if((!item->value.empty()) && ((TASK_TYPE_SET_CMD_RULE == dev_reqs->type) || (TASK_TYPE_LINKAGE == dev_reqs->type))){                
                resp_data.value = get_real_value_str(modbus_control, item->value, vstr);
                r = write_handler(cm, channel_ptr, mm, dev, modbus_control, item->value);
            } else {
                // resp_data.value = modbus_control->def_value;
                resp_data.value = get_real_value_str(modbus_control, modbus_control->def_value, vstr);
                r = write_handler(cm, channel_ptr, mm, dev, modbus_control, modbus_control->def_value);
            }
        }
        
        if(0 == r){
            resp_data.msg  = "命令执行成功";
            resp_data.code = 0;
        } else {
            resp_data.msg  = "命令执行失败";
            resp_data.code = -1;
        }
        for(auto p = cm->get_modbus_points()->begin(); p != cm->get_modbus_points()->end(); p++){
            if(p->second.reg_addr == resp_data.reg_addr){
                resp_data.name = p->second.name;
                resp_data.data_type = p->second.data_type;
                resp_update_datas.insert(std::make_pair(resp_data.name, resp_data));
                break;
            }
        }
        //保存错误信息
        resp_datas.insert(std::make_pair(resp_data.name, resp_data));
    }
    //上报命令执行结果
    if(TASK_TYPE_SET_CMD_RULE == dev_reqs->type){
        if(0 > dev_reqs->req_id){
            dev_reqs->req_id = get_req_id();
        }
        if(resp_datas.empty()){
            CReportManager::getInstance().cmd_response(dev_reqs->dev_id, dev_reqs->req_id, "CMD_SET_COMMAND_RULE", -1, "命令未执行");
        } else {
            CReportManager::getInstance().push_data(dev->id, resp_update_datas);
            CReportManager::getInstance().cmd_response(dev_reqs->dev_id, dev_reqs->req_id, "CMD_SET_COMMAND_RULE", resp_datas.begin()->second.code, resp_datas.begin()->second.msg);
        }
    }
    return ret;
}
//此方法还可以优化
int MessageManager::run(std::shared_ptr<CConfigManager> cm, IChannel *channel_ptr, std::shared_ptr<CMosquittoManager> mm, IModule &im, TASK_PARAMS *dev_reqs){
    if((NULL == cm) || (nullptr == channel_ptr) || (nullptr == dev_reqs)) {
        return -1;
    }
    auto dev = cm->get_terminal_device(dev_reqs->dev_id);
    if(nullptr == dev){
        CLogger::getInstance()->log_w("不存在该设备:%lld", dev_reqs->dev_id);
        return -1;
    }
    if(1ll > dev->parent_id){
        return no_parent_dev_handle(cm, channel_ptr, mm, im, dev_reqs, dev);
    } else {
        return parent_dev_handle(cm, channel_ptr, mm, im, dev_reqs, dev);
    }
}
//
int MessageManager::save_interaction_log(int64_t dev_id, int64_t task_id, uint8_t *req, int req_len, uint8_t *rep, int rep_len){
    DEV_INTERACTIVE_MSG interfacive_msg;
    interfacive_msg.dev_id = dev_id;
    save_log(req, req_len, INITIATE_REQUEST);
    save_log(rep, rep_len, INITIATE_RESPONSE);
    CDBManager::getInstance()->insert_log(interfacive_msg);
    return 0;
}
//
int MessageManager::get_write_request_msg(int slave_id, int reg_addr, int nb, uint8_t *data, int func_code, uint8_t *buf, int &len){
    int ret = -1;
    if((NULL == data) || (NULL == buf) || (1 > len)){
        return -1;
    }
    try{
        switch(func_code){
            case MODBUS_FC_WRITE_SINGLE_COIL:
                ret = get_modbus_write_bit_msg(slave_id, reg_addr, data[0], buf, len);
            break;
            case MODBUS_FC_WRITE_SINGLE_REGISTER:
                ret = get_modbus_write_register_msg(slave_id, reg_addr, *((uint16_t *)data), buf, len);
            break;
            case MODBUS_FC_WRITE_MULTIPLE_COILS:
                ret = get_modbus_write_bits_msg(slave_id, reg_addr, nb, data, buf, len);
            break;
            case MODBUS_FC_WRITE_MULTIPLE_REGISTERS:
                ret = get_modbus_write_registers_msg(slave_id, reg_addr, nb, (uint16_t *)data, buf, len);
            break;
            default:
            break;
        }
        if(0 < ret){
            CLogger::getInstance()->log_buf((char *)buf, ret);
        } else {
            CLogger::getInstance()->log_i("modbus write func_code: %d.\n", func_code);
        }
    }catch(...){
        printf("写数据异常。\n");
    }
    return ret;
}
int MessageManager::get_read_request_msg(int slave_id, int reg_addr, int reg_count, int func_code, uint8_t *buf, int &len){
    int ret = -1;
    try{
        switch(func_code){
            case MODBUS_FC_READ_COILS:
                ret = get_modbus_read_bits_msg(slave_id, reg_addr, reg_count, buf, len);
            break;
            case MODBUS_FC_READ_DISCRETE_INPUTS:
                ret = get_modbus_read_input_bits_msg(slave_id, reg_addr, reg_count, buf, len);
            break;
            case MODBUS_FC_READ_HOLDING_REGISTERS:
                ret = get_modbus_read_registers_msg(slave_id, reg_addr, reg_count, buf, len);
            break;
            case MODBUS_FC_READ_INPUT_REGISTERS:
                ret = get_modbus_read_input_registers_msg(slave_id, reg_addr, reg_count, buf, len);
            break;
            default:
            break;
        }
        if(0 < ret){
            len = ret;
            CLogger::getInstance()->LOG_I("Request Data:");
            CLogger::getInstance()->LOG_BUF((char *)buf, ret);
        } else {
            len = 0;
            CLogger::getInstance()->LOG_I("modbus read func_code: %d.\n", func_code);
        }
    }catch(...){
        printf("读取数据异常。\n");
    }
    return ret;
}
void MessageManager::get_write_data_by_order(std::string byte_order, uint16_t *data){
    std::string strUpper = "";
    strUpper.resize(byte_order.length());
    transform(byte_order.begin(), byte_order.end(), strUpper.begin(), ::toupper);
    if(0 == byte_order.compare("1234")){
    } else if(0 == byte_order.compare("2143")){
        bswap_16((uint8_t *)data);
        bswap_16((uint8_t *)(data+1));
    } else if(0 == byte_order.compare("3412")){
        wswap_32(data);
    } else if(0 == byte_order.compare("4321")){
        wswap_32(data);
        bswap_16((uint8_t *)data);
        bswap_16((uint8_t *)(data+1));
    } else if(0 == byte_order.compare("12")){
    } else if(0 == byte_order.compare("21")){
        bswap_16((uint8_t *)data);
    }
}
int MessageManager::get_write_data(std::shared_ptr<MODBUS_COMMAND> modbus_control, std::string value, uint8_t *data, int *len){
    if((nullptr == modbus_control) || value.empty() || (NULL == data) || (NULL == len)){
        return 0;
    }
    try{
        switch(modbus_control->data_type){
            case DATA_TYPE_BIT:
                data[0] = (uint8_t)std::stoi(value);
                *len = 1;
            break;
            case DATA_TYPE_BITS:
                str_to_hex(value.c_str(), data, len);
            break;
            case DATA_TYPE_SHORT:
                *(short *)data = get_real_value<short>(modbus_control->formula, modbus_control->scale, modbus_control->decimal, value);
                *len = 1;
            break;
            case DATA_TYPE_UNSIGNED_SHORT:
                *(uint16_t *)data = get_real_value<uint16_t>(modbus_control->formula, modbus_control->scale, modbus_control->decimal, value);
                *len = 1;
            break;
            case DATA_TYPE_INT:
                *(int *)data = get_real_value<int>(modbus_control->formula, modbus_control->scale, modbus_control->decimal, value);
                *len = 2;
            break;
            case DATA_TYPE_UNSIGNED_INT:
                *(uint32_t *)data = get_real_value<uint32_t>(modbus_control->formula, modbus_control->scale, modbus_control->decimal, value);
                *len = 2;
            break;
            case DATA_TYPE_FLOAT:
                *(float *)data = get_real_value<float>(modbus_control->formula, modbus_control->scale, modbus_control->decimal, value);
                *len = 2;
            break;
            default:
                *len = 0;
            break;
        }
    } catch(...){
        CLogger::getInstance()->log_e("获取写数据异常.");
        *len = 0;
    }
    return *len;
}
int MessageManager::gw_write_handler(std::shared_ptr<MODBUS_COMMAND> modbus_control, std::string &value){
    int data_len = 2048;
    int ret = -1;
    uint8_t data[data_len] = {0};
    
    if(1 > get_write_data(modbus_control, value, data, &data_len)){
        CLogger::getInstance()->log_w("网关待写数据格式不正确:%s", value.c_str());
        return -1;
    }
    modbus_control->reg_count = data_len;
    if((modbus_control->data_type != DATA_TYPE_BIT) || (modbus_control->data_type != DATA_TYPE_BITS)){
        get_write_data_by_order(modbus_control->byte_order, (uint16_t *)data);
    }
    //
    try{
        switch(modbus_control->func_code){
            case MODBUS_FC_WRITE_SINGLE_COIL:
                ret = gw_write_coil(modbus_control->reg_addr, data[0]);
            break;
            case MODBUS_FC_WRITE_SINGLE_REGISTER:
                ret = gw_write_register(modbus_control->reg_addr, *((uint16_t *)data));
            break;
            case MODBUS_FC_WRITE_MULTIPLE_COILS:
                ret = gw_write_coils(modbus_control->reg_addr, modbus_control->reg_count, data);
            break;
            case MODBUS_FC_WRITE_MULTIPLE_REGISTERS:
                ret = gw_write_registers(modbus_control->reg_addr, modbus_control->reg_count, (uint16_t *)data);
            break;
            default:
            break;
        }
    }catch(...){
        printf("网关设置: 写数据异常。\n");
    }

    return 0;

}

int MessageManager::write_handler(std::shared_ptr<CConfigManager> cm, 
                                IChannel *channel_ptr, 
                                std::shared_ptr<CMosquittoManager> mm,
                                std::shared_ptr<TERMINAL_DEVICE> dev, 
                                std::shared_ptr<MODBUS_COMMAND> modbus_control, 
                                std::string &value){
    int req_len = MODBUS_MAX_ADU_LENGTH;
    uint8_t req[req_len] = {0};
    uint8_t tmp_rep[MODBUS_MAX_ADU_LENGTH] = {0};
    int tmp_rep_len = MODBUS_MAX_ADU_LENGTH;
    int data_len = 2048;
    uint8_t data[data_len] = {0};
    auto proto = cm->get_protocol_rule(modbus_control->protocol_id);
    if(nullptr == proto){
        return -1;
    }
    if(1 > get_write_data(modbus_control, value, data, &data_len)){
        CLogger::getInstance()->log_w("待写数据格式不正确:%s", value.c_str());
        return -1;
    }
    modbus_control->reg_count = data_len;
    if((modbus_control->data_type != DATA_TYPE_BIT) || (modbus_control->data_type != DATA_TYPE_BITS)){
        get_write_data_by_order(modbus_control->byte_order, (uint16_t *)data);
    }
    int ret = get_write_request_msg(atoi(dev->slave_addr.c_str()), 
                                    modbus_control->reg_addr, 
                                    modbus_control->reg_count, 
                                    data,  
                                    modbus_control->func_code, 
                                    req, req_len);
    
    if(0 > channel_ptr->send_msg(req, ret, tmp_rep, tmp_rep_len, proto->timeout)){
        return -1;
    }else{
        if(1 > check_confirmation(req, tmp_rep, tmp_rep_len)){
            // if(0x80 < tmp_rep[1]){
                CLogger::getInstance()->log_w("设置数据参数失败");
                return -1;
            // }
        }
    }
    //保存交互日志
    return 0;
}
int MessageManager::get_bit_value(uint8_t *rsp, uint8_t *data, int nb){
    if((NULL == rsp) || (NULL == data) || (1 > nb)){
        return -1;
    }
    int offset = 3;
    int n = *(uint8_t *)(rsp+2);
    int pos = 0;
    for (int i = 0; i < n; i++) {
        /* Shift reg hi_byte to temp */
        uint8_t temp = rsp[offset+i];

        for (uint8_t bit = 0x01; (bit & 0xff) && (pos < nb);) {
            data[pos++] = (temp & bit) ? TRUE : FALSE;
            bit = bit << 1;
        }
    }
    return pos;
}
int MessageManager::gw_collect(int reg_addr, int nb, int func_code, uint8_t *buf, int &len){
    int ret = -1;
    try{
        switch(func_code){
            case MODBUS_FC_WRITE_SINGLE_COIL:
                ret = gw_read_coil(reg_addr, buf[0]);
            break;
            case MODBUS_FC_WRITE_SINGLE_REGISTER:
                ret = gw_read_register(reg_addr, *((uint16_t *)buf));
            break;
            case MODBUS_FC_WRITE_MULTIPLE_COILS:
                ret = gw_read_coils(reg_addr, nb, buf);
            break;
            case MODBUS_FC_WRITE_MULTIPLE_REGISTERS:
                ret = gw_read_registers(reg_addr, nb, (uint16_t *)buf);
            break;
            default:
            break;
        }
    }catch(...){
        printf("网关设置: 读数据异常。\n");
    }
    return ret;
}

int MessageManager::collect(IChannel *channel_ptr, 
                            int64_t dev_id, 
                            int64_t task_id,            //暂时没用可删除
                            int slave_id, 
                            int reg_addr, 
                            int reg_count, 
                            int func_code, 
                            uint8_t *buf, 
                            int &len,
                            int timeout){
    int req_len = 32;
    uint8_t req[req_len] = {0};
    uint8_t rep[MODBUS_MAX_ADU_LENGTH] = {0};
    int rep_len = MODBUS_MAX_ADU_LENGTH;

    //临时添加,支持本地modbus
    if(0xFF == slave_id){
        len = 1;
        gw_collect(reg_addr, reg_count, func_code, buf, len);
        return len;
    }

    int ret = get_read_request_msg(slave_id, reg_addr, reg_count, func_code, req, req_len);
    if(0 < ret){
        ret = channel_ptr->send_msg(req, ret, rep, rep_len, timeout);
        if((0 < ret) && (MODBUS_MAX_ADU_LENGTH != rep_len)){
            CLogger::getInstance()->LOG_I("Response Data:");
            CLogger::getInstance()->LOG_BUF((char*)rep,rep_len);
            
            //保存交互日志
            save_interaction_log(dev_id, task_id, req, req_len, rep , rep_len);
            
            if(1 > check_confirmation(req, rep, rep_len)){
                // if(0x80 < rep[1]){
                    CLogger::getInstance()->LOG_W("采集数据返回失败(slave_id: %lld)", slave_id);
                    return -1;
                // }
            }
            
            if((0x01 == func_code) || (0x02 == func_code)){
                len = get_bit_value(rep, buf, reg_count);
                if(1 > len){
                    CLogger::getInstance()->LOG_W("采集开关量数据失败");
                    return -2;
                }
            } else {
                len = modbus_get_data_len(rep);
                if(1 > len){
                    CLogger::getInstance()->LOG_W("采集信号量数据失败");
                    return -3;
                }
                memcpy(buf, rep+3, len);
            }
        } else {
            CLogger::getInstance()->LOG_W("采集数据超时");
            save_interaction_log(dev_id, task_id, req, req_len, NULL , -1);
            return -200;
        }
    } else {
        CLogger::getInstance()->LOG_W("采集数据组包失败");
        return -10;
    }
    return len;
}
/* 可优化 */
int MessageManager::read_handler(std::shared_ptr<CConfigManager> cm, 
                                IChannel *channel_ptr, 
                                std::shared_ptr<CMosquittoManager> mm,
                                std::shared_ptr<TERMINAL_DEVICE> dev, 
                                int64_t task_id, 
                                int64_t task_item_id, 
                                std::shared_ptr<COLLECT_DATA> collect_data){
    
    uint8_t tmp_rep[MODBUS_MAX_ADU_LENGTH] = {0};
    int tmp_rep_len = MODBUS_MAX_ADU_LENGTH;
    int n = 0;
    int offset = 0;
    int rep_len = MAX_REPONSE_LEN;
    int offset_addr = 0;

    auto modbus_request = cm->get_modbus_request(task_item_id);
    if(nullptr == modbus_request){
        CLogger::getInstance()->log_w("不存在该请求项:%lld", task_item_id);
        return -1;
    }
    auto proto = cm->get_protocol_rule(modbus_request->protocol_id);
    if(nullptr == proto){
        CLogger::getInstance()->log_w("不存在该协议:%s", proto->protocol_name.c_str());
        return -1;
    }

    uint8_t *rep = (uint8_t *)malloc(rep_len);
    int m = modbus_request->reg_count%MODBUS_MAX_WR_REGISTERS;
    if((0x01 == modbus_request->func_code) || (0x02 == modbus_request->func_code)){
        n = modbus_request->reg_count/MODBUS_MAX_READ_BITS;
        m = modbus_request->reg_count%MODBUS_MAX_READ_BITS;
    } else if((0x03 == modbus_request->func_code) || (0x04 == modbus_request->func_code)){
        n = modbus_request->reg_count/MODBUS_MAX_WR_REGISTERS;
        m = modbus_request->reg_count%MODBUS_MAX_WR_REGISTERS;
    }
    int slave_id = atoi(dev->slave_addr.c_str());
    for(int i = 0; i < n; i++){
        tmp_rep_len = MODBUS_MAX_ADU_LENGTH;
        offset_addr = i*MODBUS_MAX_WR_REGISTERS;

        if(0 < collect(channel_ptr, dev->id, task_id, slave_id, modbus_request->reg_addr+offset_addr, MODBUS_MAX_WR_REGISTERS, modbus_request->func_code, tmp_rep, tmp_rep_len, proto->timeout)){
            if(tmp_rep_len + offset > rep_len){
                rep_len += MAX_REPONSE_LEN;
                uint8_t *p = (uint8_t *)realloc(rep, rep_len);
                if(NULL != p){
                    rep = p;
                }                
            }
            memcpy(rep+offset, tmp_rep, tmp_rep_len);
            offset += tmp_rep_len;
        }
    }
    if(0 < m){
        tmp_rep_len = MODBUS_MAX_ADU_LENGTH;
        if(0 < collect(channel_ptr, dev->id, task_id, slave_id, modbus_request->reg_addr+offset_addr, m, modbus_request->func_code, tmp_rep, tmp_rep_len, proto->timeout)){
            if(tmp_rep_len + offset > rep_len){
                rep_len += MAX_REPONSE_LEN;
                uint8_t *p = (uint8_t *)realloc(rep, rep_len);
                if(NULL != p){
                    rep = p;
                }
            }
            memcpy(rep+offset, tmp_rep, tmp_rep_len);
            offset += tmp_rep_len;
        }
    }
    //数据按照请求数保存数据
    if(0 < offset){
        collect_data->dev_id = dev->id;
        collect_data->protocol_id = dev->protocol_id;
        collect_data->reg_count = modbus_request->reg_count;
        collect_data->start_addr = modbus_request->reg_addr;
        collect_data->data = rep;
        collect_data->data_len = offset;
        collect_data->func_code = modbus_request->func_code;
    } else {
        if(NULL != rep){
            free(rep);
            rep = NULL;
        }
        return -1;
    }
    
    return 0;
}
int MessageManager::get_read_func_code(int funcCode){
    switch(funcCode){
        case 0x0F:
        case 0x05:
           return 0x01;
        break;
        case 0x10:
        case 0x06:
            return 0x03;
        break;
        case 0x02:
        case 0x04:
            return funcCode;
        break;
        default:
        break;
    }
    return 0x0;
}
//
int MessageManager::read_cmd_handler(IChannel *channel_ptr, 
                                std::shared_ptr<TERMINAL_DEVICE> dev, 
                                std::shared_ptr<MODBUS_COMMAND> command, 
                                std::shared_ptr<COLLECT_DATA> collect_data,
                                int timeout){
    int len = 0;
    int rep_len = 64;
    // uint8_t rep[rep_len] = {0};
    uint8_t *rep = (uint8_t *)malloc(rep_len);
    int func_code = get_read_func_code(command->func_code);
    if(0 == func_code){
        CLogger::getInstance()->log_w("modbus读指令(未知功能码)");
        return -201;
    }
    len = collect(channel_ptr, dev->id, -1, atoi(dev->slave_addr.c_str()), command->reg_addr, command->reg_count, func_code, rep, rep_len, timeout);
    if(0 < len){
        //数据按照请求数保存数据
        collect_data->name = command->name;
        collect_data->dev_id = dev->id;
        collect_data->protocol_id = dev->protocol_id;
        collect_data->reg_count = command->reg_count;
        collect_data->start_addr = command->reg_addr;
        collect_data->data = rep;
        collect_data->data_len = len;
        collect_data->func_code = func_code;
    } else {
        if(NULL != rep){
            free(rep);
            rep = NULL;
        }
        return len;
    }
    
    return 0;
}

int MessageManager::write_request_frame(int slave_id, std::shared_ptr<MODBUS_COMMAND> modbus_control, std::string &value, uint8_t *frame, int frame_len){
    int data_len = MODBUS_MAX_ADU_LENGTH;
    uint8_t data[data_len] = {0};

    if((NULL == frame) || (0 >= frame_len)){
        return -1;
    }
    
    if(1 > get_write_data(modbus_control, value, data, &data_len)){
        CLogger::getInstance()->log_w("待写数据格式不正确:%s", value.c_str());
        return -1;
    }
    modbus_control->reg_count = data_len;
    if((modbus_control->data_type != DATA_TYPE_BIT) || (modbus_control->data_type != DATA_TYPE_BITS)){
        get_write_data_by_order(modbus_control->byte_order, (uint16_t *)data);
    }
    int ret = get_write_request_msg(slave_id, 
                                    modbus_control->reg_addr, 
                                    modbus_control->reg_count, 
                                    data,  
                                    modbus_control->func_code, 
                                    frame, frame_len);
    return ret;
}

int MessageManager::read_request_frame(int slave_id, std::shared_ptr<MODBUS_REQUEST> modbus_request, uint8_t *frame, int frame_len){
    if((0 >= slave_id) || (NULL == frame) || (0 >= frame_len)){
        return -1;
    }
    return get_read_request_msg(slave_id, modbus_request->reg_addr, modbus_request->reg_count, modbus_request->func_code, frame, frame_len);
}

std::shared_ptr<TERMINAL_DEVICE> MessageManager::get_parent_dev(std::shared_ptr<CConfigManager> cm, uint64_t parent_id){
    
    return cm->get_terminal_device(parent_id);
}
#if 0
int MessageManager::run(std::shared_ptr<CConfigManager> cm, IChannel *channel_ptr, std::shared_ptr<CMosquittoManager> mm, IModule &im, TASK_PARAMS *dev_reqs){
    if((NULL == cm) || (nullptr == channel_ptr) || (nullptr == dev_reqs)) {
        return -1;
    }
    auto dev = cm->get_terminal_device(dev_reqs->dev_id);
    if(nullptr == dev){
        CLogger::getInstance()->log_w("不存在该设备:%lld", dev_reqs->dev_id);
        return -1;
    }
    int ret = 0;
    switch(dev_reqs->type){
        case TASK_TYPE_GET_CMD_RULE:
            ret = run_read_cmd(cm, channel_ptr, mm, dev_reqs);
        break;
        case TASK_TYPE_COLLECT:
            ret = run_collect(cm, channel_ptr, mm, module_func, dev_reqs, dev);
        break;
        case TASK_TYPE_CTL_CMD_RULE:
        case TASK_TYPE_SET_CMD_RULE:
        case TASK_TYPE_LINKAGE:
            ret = run_write_cmd(cm, channel_ptr, mm, dev_reqs);
        break;
        default:
            ret = -1;
            CLogger::getInstance()->log_w("不存在的任务类型");
        break;
    }
    return ret;
}
#endif
int MessageManager::no_parent_dev_handle(std::shared_ptr<CConfigManager> cm, 
                                      IChannel *channel_ptr, 
                                      std::shared_ptr<CMosquittoManager> mm, 
                                      IModule &im, 
                                      TASK_PARAMS *dev_reqs, 
                                      std::shared_ptr<TERMINAL_DEVICE> dev){
    int ret = 0;
    switch(dev_reqs->type){
        case TASK_TYPE_GET_CMD_RULE:
            ret = run_read_cmd(cm, channel_ptr, mm, dev_reqs, dev);
        break;
        case TASK_TYPE_COLLECT:
            ret = run_collect(cm, channel_ptr, mm, dev_reqs, dev);
        break;
        case TASK_TYPE_CTL_CMD_RULE:
        case TASK_TYPE_SET_CMD_RULE:
        case TASK_TYPE_LINKAGE:
            ret = run_write_cmd(cm, channel_ptr, mm, dev_reqs, dev);
        break;
        default:
            ret = -1;
            CLogger::getInstance()->log_w("不存在的任务类型");
        break;
    }
    return ret;
}
int MessageManager::package_parent_frame(std::shared_ptr<CConfigManager> cm, IModule &im, stack<void *> &parsers, std::shared_ptr<TERMINAL_DEVICE> pdev, uint8_t *data, int &len){
    if((nullptr == cm) || (nullptr == pdev) || (NULL == data) || (1 > len)){
        return len;
    }
    if(0ll < pdev->parent_id){
        auto ppdev = cm->get_terminal_device(pdev->parent_id);
        if(ppdev == nullptr){
            return len;
        }
        auto pproto = cm->get_protocol_rule(ppdev->protocol_id);
        if(pproto == nullptr){
            return len;
        }
        TFUNCTION_PACKET_PARENT_FRAME func_packet = (TFUNCTION_PACKET_PARENT_FRAME)IModule::getInstance().get_func(std::to_string(pproto->protocol_type), FUNC_PACKET_PARENT_FRAME_NAME);
        if(NULL == func_packet){
            return len;
        }
        auto func_pareser = (TFUNCTION_PARSE_PARENT_FRAME)IModule::getInstance().get_func(std::to_string(pproto->protocol_type), FUNC_PARSE_PARENT_FRAME_NAME);
        if(NULL == func_pareser){
            return len;
        }
        parsers.push((void *)func_pareser);
        func_packet(cm, ppdev, data, len);
        return package_parent_frame(cm, im, parsers, ppdev, data, len);
    } else {
        return len;
    }
}
int MessageManager::parse_parent_frame(std::shared_ptr<CConfigManager> cm, stack<void *> &parsers, uint8_t *data, int &len){
    while(!parsers.empty()){
        TFUNCTION_PARSE_PARENT_FRAME func_parse = (TFUNCTION_PARSE_PARENT_FRAME)parsers.top();
        parsers.pop();
        func_parse(cm, data, len);
    }

    return len;
}
int MessageManager::parse_modbus_frame(uint8_t *frame, int frame_len, int func_code, uint8_t *target, int &target_len){
    if(2 <= frame_len){
        if(0x80 < frame[1]){
            CLogger::getInstance()->log_w("采集数据返回失败(%02X)", frame[0]);
            return -1;
        }
    }
    if((0x01 == func_code) || (0x02 == func_code)){
        target_len = get_bit_value(frame, target, MODBUS_MAX_WR_REGISTERS);
        if(1 > target_len){
            CLogger::getInstance()->log_w("采集开关量数据失败");
            return -2;
        }
    } else {
        target_len = modbus_get_data_len(frame);
        if(1 > target_len){
            CLogger::getInstance()->log_w("采集信号量数据失败");
            return -3;
        }
        memcpy(target, frame+3, target_len);
    }
    return target_len;
}

int MessageManager::parent_run_read_cmd(std::shared_ptr<CConfigManager> cm, 
                                      IChannel *channel_ptr, 
                                      std::shared_ptr<CMosquittoManager> mm, 
                                      IModule &im, 
                                      TASK_PARAMS *dev_reqs, 
                                      std::shared_ptr<TERMINAL_DEVICE> dev){
    if(nullptr == cm || (NULL == channel_ptr) || (nullptr == mm) || (NULL == dev_reqs) || (nullptr == dev)){
        return -1;
    }
    int ret = -1;
    int req_len = 128;
    int rep_len = 2048;
    int target_len = MAX_REPONSE_LEN;
    uint8_t req[req_len] = {0};
    uint8_t rep[rep_len] = {0};
    uint8_t *target = (uint8_t *)malloc(target_len);
    std::list<std::shared_ptr<COLLECT_DATA>> data;
    std::unordered_map<std::string, RESPONSE_DATA> resp_datas;
    for(auto item = dev_reqs->items.begin(); item != dev_reqs->items.end(); item++){
        stack<void *> parsers;
        auto modbus_command = cm->get_modbus_command(item->item_id);
        RESPONSE_DATA resp_data;
        resp_data.point_id = -1;
        resp_data.code  = -1;
        resp_data.name  = std::to_string(item->item_id);
        resp_data.msg   = "";
        resp_data.value = "";
        if(nullptr == modbus_command){
            CLogger::getInstance()->log_w("不存在该读取指令:%lld", item->item_id);
            resp_data.msg = "不存在该读取指令";
            resp_datas.insert(std::make_pair(resp_data.name, resp_data));
            continue;
        }
        auto proto = cm->get_protocol_rule(modbus_command->protocol_id);
        if(nullptr == proto){
            continue;
        }
        int func_code = get_read_func_code(modbus_command->func_code);
        ret = get_read_request_msg(atoi(dev->slave_addr.c_str()), modbus_command->reg_addr, modbus_command->reg_count, modbus_command->func_code, req, req_len);
        if(0 > ret){
            resp_data.msg = "读取请求帧封装失败";
            resp_datas.insert(std::make_pair(resp_data.name, resp_data));
            continue;
        }
        //封装父级帧
        package_parent_frame(cm, im, parsers, dev, req, req_len);
        //发送请求
        if(0 > channel_ptr->send_msg(req, req_len, rep, rep_len, proto->timeout)){
            resp_data.msg = "请求数据失败";
            resp_datas.insert(std::make_pair(resp_data.name, resp_data));
            continue;
        }else{
            //保存交互日志
            save_interaction_log(dev->id, dev_reqs->task_id, req, req_len, rep , rep_len);
            if(0 < check_confirmation(req, rep, rep_len)){
                // if(0x80 < rep[1]){
                //     CLogger::getInstance()->log_w("请求数据失败应答");
                //     resp_data.msg = "请求数据失败应答";
                //     resp_datas.insert(std::make_pair(resp_data.name, resp_data));
                //     continue;
                // }
                //解包父级帧
                parse_parent_frame(cm, parsers, rep, rep_len);
                //解包modbus帧
                parse_modbus_frame(rep, rep_len, modbus_command->func_code, target, target_len);
                std::shared_ptr<COLLECT_DATA> collect_data = std::make_shared<COLLECT_DATA>();
                collect_data->dev_id = dev->id;
                collect_data->protocol_id = dev->protocol_id;
                collect_data->reg_count = modbus_command->reg_count;
                collect_data->start_addr = modbus_command->reg_addr;
                collect_data->data = target;
                collect_data->data_len = target_len;
                collect_data->func_code = func_code;
                data.push_back(collect_data);
            } else {
                CLogger::getInstance()->log_w("请求数据应答异常");
                resp_data.msg = "请求数据应答异常";
                resp_datas.insert(std::make_pair(resp_data.name, resp_data));
            }
        }
        // get_read_request_msg(dev->slave_addr, modbus_command->reg_addr, modbus_command->reg_count, func_code, req, req_len);
    }
    if(0 < data.size()){
        ModbusParseTask *pTask = new ModbusParseTask(cm, mm, dev_reqs, data, resp_datas);
        c_thread_pool.add_task(pTask);
    }
    return 0;
}
/**
 * @brief 临时调试，后续需要优化
*/
int MessageManager::parent_run_collect(std::shared_ptr<CConfigManager> cm, 
                                      IChannel *channel_ptr, 
                                      std::shared_ptr<CMosquittoManager> mm, 
                                      IModule &im, 
                                      TASK_PARAMS *dev_reqs, 
                                      std::shared_ptr<TERMINAL_DEVICE> dev){
    if(nullptr == cm || (NULL == channel_ptr) || (nullptr == mm) || (NULL == dev_reqs) || (nullptr == dev)){
        return -1;
    }
    int ret = 0;
    int req_len = 128;
    // int save_req_len = 128;
    int rep_len = MAX_REPONSE_LEN;
    int target_len = MAX_REPONSE_LEN;
    uint8_t req[req_len] = {0};
    uint8_t rep[rep_len] = {0};
    uint8_t save_req[req_len] = {0};

    std::list<std::shared_ptr<COLLECT_DATA>> data;
    std::unordered_map<std::string, RESPONSE_DATA> resp_datas;
    for(auto item = dev_reqs->items.begin(); item != dev_reqs->items.end(); item++){
        auto task_item = cm->get_item(item->item_id);
        RESPONSE_DATA resp_data;
        resp_data.point_id = -1;
        resp_data.code  = -1;
        resp_data.name  = std::to_string(item->item_id);
        resp_data.msg   = "";
        resp_data.value = "";
        if(nullptr == task_item){
            CLogger::getInstance()->log_w("不存在该任务项:%lld", item->item_id);
            resp_data.msg = "不存在该读取项";
            resp_datas.insert(std::make_pair(resp_data.name, resp_data));
            continue;
        }
        req_len = 128;
        rep_len = MAX_REPONSE_LEN;
        target_len = MAX_REPONSE_LEN;
        uint8_t *target = (uint8_t *)malloc(target_len);
        stack<void *> parsers;
        auto modbus_request = cm->get_modbus_request(task_item->task_item_id);
        if(modbus_request != nullptr){
            auto proto = cm->get_protocol_rule(modbus_request->protocol_id);
            if(nullptr == proto){
                continue;
            }
            ret = get_read_request_msg(atoi(dev->slave_addr.c_str()), modbus_request->reg_addr, modbus_request->reg_count, modbus_request->func_code, req, req_len);
            if(0 > ret){
                resp_data.msg = "读取请求帧封装失败";
                resp_datas.insert(std::make_pair(resp_data.name, resp_data));
                continue;
            }
            memcpy(save_req, req, req_len);
            // int save_req_len = req_len;
            //封装父级帧
            package_parent_frame(cm, im, parsers, dev, req, req_len);
            if (0ll < dev->parent_id){
                auto pdev = cm->get_terminal_device(dev->parent_id);
                if (pdev){
                    channel_ptr->set_slave_addr(pdev->slave_addr);
                }
            }
            
            //发送请求
            if(0 > channel_ptr->send_msg(req, req_len, rep, rep_len, proto->timeout)){
                resp_data.msg = "请求数据失败";
                resp_datas.insert(std::make_pair(resp_data.name, resp_data));
                continue;
            }else{
                //保存交互日志
                save_interaction_log(dev->id, dev_reqs->task_id, req, req_len, rep , rep_len);
                //解包父级帧
                parse_parent_frame(cm, parsers, rep, rep_len);
                // CLogger::getInstance()->log_w("modbus request and response");
                // CLogger::getInstance()->log_buf((const char*)save_req, save_req_len);
                // CLogger::getInstance()->log_i("======================================");
                // CLogger::getInstance()->log_buf((const char *)rep, rep_len);
                if(0 < check_confirmation(save_req, rep, rep_len)){
                    // if(0x80 < rep[1]){
                    //     CLogger::getInstance()->log_w("请求数据失败应答");
                    //     resp_data.msg = "请求数据失败应答";
                    //     resp_datas.insert(std::make_pair(resp_data.name, resp_data));
                    //     continue;
                    // }
                    
                    //解包modbus帧
                    parse_modbus_frame(rep, rep_len, modbus_request->func_code, target, target_len);
                    std::shared_ptr<COLLECT_DATA> collect_data = std::make_shared<COLLECT_DATA>();
                    collect_data->dev_id = dev->id;
                    collect_data->protocol_id = dev->protocol_id;
                    collect_data->reg_count = modbus_request->reg_count;
                    collect_data->start_addr = modbus_request->reg_addr;
                    collect_data->data = target;
                    collect_data->data_len = target_len;
                    collect_data->func_code = modbus_request->func_code;
                    data.push_back(collect_data);
                } else {
                    CLogger::getInstance()->log_w("请求数据应答异常");
                    resp_data.msg = "请求数据应答异常";
                    resp_datas.insert(std::make_pair(resp_data.name, resp_data));
                }
            }
        } else {
            CLogger::getInstance()->log_w("不存在该请求规则:%lld", task_item->task_item_id);
            resp_data.msg = "不存在该请求规则";
            resp_datas.insert(std::make_pair(resp_data.name, resp_data));
        }
    }
    if(0 < data.size()){
        ModbusParseTask *pTask = new ModbusParseTask(cm, mm, dev_reqs, data, resp_datas);
        c_thread_pool.add_task(pTask);
    }
    return ret;
}
int MessageManager::parent_run_write_cmd(std::shared_ptr<CConfigManager> cm, 
                                      IChannel *channel_ptr, 
                                      std::shared_ptr<CMosquittoManager> mm, 
                                      IModule &im, 
                                      TASK_PARAMS *dev_reqs, 
                                      std::shared_ptr<TERMINAL_DEVICE> dev){
    if(nullptr == cm || (NULL == channel_ptr) || (nullptr == mm) || (NULL == dev_reqs) || (nullptr == dev)){
        return -1;
    }
    int ret = -1;
    int req_len = 32;
    int rep_len = 2048;
    uint8_t req[req_len] = {0};
    uint8_t rep[rep_len] = {0};
    uint8_t save_req[req_len] = {0};
    std::unordered_map<std::string, RESPONSE_DATA> resp_datas;
    for(auto item = dev_reqs->items.begin(); item != dev_reqs->items.end(); item++){
        ret = -1;
        stack<void *> parsers;
        RESPONSE_DATA resp_data;
        resp_data.name = dev->name;
        resp_data.value = "";
        auto modbus_command = cm->get_modbus_command(item->item_id);
        if(nullptr == modbus_command){
            CLogger::getInstance()->log_w("不存在该指令规则:%lld", item->item_id);
            resp_data.msg  = "命令执行失败，不存在该指令规则";
            resp_data.code = -1;
            resp_data.point_id = -1;
            resp_datas.insert(std::make_pair(modbus_command->name, resp_data));
            continue;
        }
        auto proto = cm->get_protocol_rule(modbus_command->protocol_id);
        if(nullptr == proto){
            continue;
        }
        resp_data.point_id = modbus_command->id;
        if(TASK_TYPE_SET_CMD_RULE == dev_reqs->type){
            req_len = write_request_frame(atoi(dev->slave_addr.c_str()), modbus_command, item->value, req, req_len);
        } else {
            req_len = write_request_frame(atoi(dev->slave_addr.c_str()), modbus_command, modbus_command->def_value, req, req_len);
        }
        memcpy(save_req, req, req_len);
        CLogger::getInstance()->log_buf((const char *)save_req, req_len);
        //封装父级帧
        package_parent_frame(cm, im, parsers, dev, req, req_len);
        CLogger::getInstance()->log_buf((const char *)req, req_len);
        if(0 > channel_ptr->send_msg(req, req_len, rep, rep_len, modbus_command->timeout)){
            resp_data.msg  = "命令执行失败，帧发送失败";
            resp_data.code = -1;
            resp_datas.insert(std::make_pair(modbus_command->name, resp_data));
            continue;
        }else{
            save_interaction_log(dev->id, dev_reqs->task_id, req, req_len, rep , rep_len);
            parse_parent_frame(cm, parsers, rep, rep_len);
            // CLogger::getInstance()->log_buf((const char *)save_req, 6);
            CLogger::getInstance()->log_buf((const char *)rep, rep_len);
            if(1 > check_confirmation(save_req, rep, rep_len)){
                // if(0x80 < rep[1]){
                    CLogger::getInstance()->log_w("设置数据参数失败");
                    resp_data.msg  = "命令执行失败，返回错误应答";
                    resp_data.code = -1;
                    resp_datas.insert(std::make_pair(modbus_command->name, resp_data));
                    continue;
                // }
            }
        }        
        resp_data.msg  = "命令执行成功";
        resp_data.code = 0;
        resp_datas.insert(std::make_pair(modbus_command->name, resp_data));
    }
    //上报命令执行结果
    if(TASK_TYPE_SET_CMD_RULE == dev_reqs->type){
        if(0 > dev_reqs->req_id){
            dev_reqs->req_id = get_req_id();
        }
        if(resp_datas.empty()){
            CReportManager::getInstance().cmd_response(dev_reqs->dev_id, dev_reqs->req_id, "CMD_SET_COMMAND_RULE", -1, "命令未执行");
        }else{
            CReportManager::getInstance().cmd_response(dev_reqs->dev_id, dev_reqs->req_id, "CMD_SET_COMMAND_RULE", resp_datas.begin()->second.code, resp_datas.begin()->second.msg);
        }
    }
}
int MessageManager::parent_dev_handle(std::shared_ptr<CConfigManager> cm, 
                                      IChannel *channel_ptr, 
                                      std::shared_ptr<CMosquittoManager> mm, 
                                      IModule &im, 
                                      TASK_PARAMS *dev_reqs, 
                                      std::shared_ptr<TERMINAL_DEVICE> dev){

    if(nullptr == cm || (NULL == channel_ptr) || (nullptr == mm) || (NULL == dev_reqs) || (nullptr == dev)){
        return -1;
    }
    int ret = 0;
    switch(dev_reqs->type){
        case TASK_TYPE_GET_CMD_RULE:{
            parent_run_read_cmd(cm, channel_ptr, mm , im, dev_reqs, dev);
        }
        break;
        case TASK_TYPE_COLLECT:{
            parent_run_collect(cm, channel_ptr, mm , im, dev_reqs, dev);
        }
        break;
        case TASK_TYPE_CTL_CMD_RULE:
        case TASK_TYPE_SET_CMD_RULE:
        case TASK_TYPE_LINKAGE:{
            parent_run_write_cmd(cm, channel_ptr, mm , im, dev_reqs, dev);
        }
        break;
        default:
            ret = -1;
            CLogger::getInstance()->log_w("不存在的任务类型");
        break;
    }
    return 0;
}